package com.atguigu.ting.realtime.app.dwd.db;

import com.atguigu.ting.realtime.app.BaseSqlApp;
import com.atguigu.ting.realtime.common.Constant;
import com.atguigu.ting.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * @Author lzc
 * @Date 2023/4/10 15:34
 */
public class DwdTradeOrderDetail extends BaseSqlApp {
    public static void main(String[] args) {
        new DwdTradeOrderDetail().init(
            3002,
            2,
            "DwdTradeOrderDetail"
        );
    }
    
    @Override
    public void handle(StreamExecutionEnvironment env,
                       StreamTableEnvironment tEnv) {
        tEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(5));
        
        // 1. 读取 ods_db
        readOdsDb(tEnv, "DwdTradeOrderDetail");
        // 2. 读取字典表
        readBaseDic(tEnv);
        // 3. 过滤订单明细表数据
        Table orderDetail = tEnv.sqlQuery(
            "select " +
                "`data`['id'] id, " +
                "`data`['order_id'] order_id, " +
                "`data`['item_id'] item_id, " +
                "`data`['item_name'] item_name, " +
                "`data`['create_time'] create_time, " +
                "cast(`data`['item_amount'] as decimal(16,2)) split_original_amount, " +
                "pt," +
                "ts  " +
                "from ods_db " +
                "where `database`='ting' " +
                "and `table`='order_detail' " +
                "and `type`='insert' ");
        tEnv.createTemporaryView("order_detail", orderDetail);
        // 4. 过滤订单表数据
        Table orderInfo = tEnv.sqlQuery(
            "select " +
                "`data`['id'] id, " +
                "`data`['user_id'] user_id, " +
                "`data`['trade_no'] trade_no, " +
                "`data`['item_type'] item_type, " +
                "cast(`data`['item_count'] as bigint) item_count " +
                "from ods_db " +
                "where `database`='ting' " +
                "and `table`='order_info' " +
                "and `type`='insert' ");
        tEnv.createTemporaryView("order_info", orderInfo);
        // 5. 过滤订单明减免表数据
        Table orderDerate = tEnv.sqlQuery(
            "select " +
                "`data`['id'] id, " +
                "`data`['order_id'] order_id, " +
                "`data`['derate_type'] derate_type, " +
                "cast(`data`['derate_amount'] as decimal(16,2)) derate_amount " +
                "from ods_db " +
                "where `database`='ting' " +
                "and `table`='order_derate' " +
                "and `type`='insert' ");
        tEnv.createTemporaryView("order_derate", orderDerate);
        
        // 6. 4 张表 join
        Table result = tEnv.sqlQuery(
            "select " +
                "oi.id order_id, " +
                "oi.user_id user_id, " +
                "od.id order_detail_id, " +
                "oi.trade_no trade_no, " +
                "od.item_id item_id, " +
                "od.item_name item_name, " +
                "oi.item_type item_type, " +
                "dic1.dic_name item_type_name, " +
                "der.derate_type derate_type, " +
                "dic2.dic_name derate_type_name, " +
                "od.split_original_amount split_original_amount, " +
                "if(der.derate_amount is null, 0, der.derate_amount) / oi.item_count  split_derate_amount, " +
                "split_original_amount - if(der.derate_amount is null, 0, der.derate_amount) / oi.item_count split_order_amount, " +
                "od.create_time create_time, " +
                "od.ts ts " +
                "from order_detail od " +
                "join order_info oi on od.order_id=oi.id " +
                "left join order_derate der on oi.id=der.order_id " +
                "join base_dic for system_time as of od.pt as dic1 " +
                "on oi.item_type=dic1.dic_code " +
                "left join base_dic for system_time as of od.pt as dic2 " +
                "on der.derate_type=dic2.dic_code"
        );
        // 7. 最终结果写出到 Kafka 中
        tEnv.executeSql(
            "create table dwd_trade_order_detail(" +
                "order_id string, " +
                "user_id string, " +
                "order_detail_id string, " +
                "trade_no string, " +
                "item_id string, " +
                "item_name string, " +
                "item_type string, " +
                "item_type_name string, " +
                "derate_type string, " +
                "derate_type_name string, " +
                "split_original_amount decimal(16,2), " +
                "split_derate_amount decimal(16,2), " +
                "split_order_amount decimal(16,2), " +
                "create_time string, " +
                "ts bigint," +
                "primary key(order_detail_id) not enforced " +
                ")" + SQLUtil.getUpsertKafkaDDL(Constant.TOPIC_DWD_TRADE_ORDER_DETAIL));
        
        result.executeInsert("dwd_trade_order_detail");
    }
}
